package org.mine.database.migration;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicLong;

public class DataTransfer {
	BlockingQueue<DataModel> queue;
	AtomicLong total = new AtomicLong();
	public AtomicLong getTotal() {
		return total;
	}

	private static int pageSize = 2000;
	
	MapiMysqlJdbcRead mysqlData = new MapiMysqlJdbcRead();
	MapiJdbcRead sqlServerData = new MapiJdbcRead();
	
	public DataTransfer(BlockingQueue<DataModel> queue) {
		this.queue = queue;
	}
	
	public AtomicLong totalCount(String tableName, String primaryKey) throws Exception{
		long lastRecordId = mysqlData.getLastRecordId(tableName, primaryKey);
		return new AtomicLong(sqlServerData.count(tableName, primaryKey, lastRecordId));
	}

	public void transfer(String tableName, String primaryKey) throws Exception {
		long lastRecordId = mysqlData.getLastRecordId(tableName, primaryKey);
		for(;;){
			long t1 = System.currentTimeMillis();
			DataModel dataModel = sqlServerData.readData(tableName, primaryKey, lastRecordId, pageSize);
			long t2 = System.currentTimeMillis();
			System.out.println("查询" + tableName + ",耗时" + (t2 -t1));
			if(dataModel != null && dataModel.getData() != null && dataModel.getData().size() > 0){
				if(dataModel.getData().get(dataModel.getData().size() -1).get(0) instanceof Integer){
					lastRecordId = (int)dataModel.getData().get(dataModel.getData().size() -1).get(0);
				}else{
					lastRecordId = (long)dataModel.getData().get(dataModel.getData().size() -1).get(0);
				}
				
				queue.put(dataModel);
				System.out.println("新增任务，当前任务数" + queue.size());
			}else{
				System.out.println(tableName + "表数据已经同步完成！");
				break;
			}
		}
		
	}

	public static void main(String[] args) throws Exception {
		String tableName = args[0];
		String primaryKey = args[1];
		int threadNum = Integer.parseInt(args[2]);
		BlockingQueue<DataModel> queue = new LinkedBlockingQueue<DataModel>(5000);
		DataTransfer dataTransfer = new DataTransfer(queue);
		AtomicLong total = dataTransfer.totalCount(tableName, primaryKey);
		for(int i = 0; i< threadNum; i++){
			MysqlInsert insertor = new MysqlInsert(queue, total);
			insertor.setName("[" + tableName + "-transfer-thread-" + i + "]");
			insertor.start();
		}
		
		dataTransfer.transfer(tableName, primaryKey);
		
	}

}
